/*
 * (C) Copyright 2014 Kurento (http://kurento.org/)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package com.one2manycall.service.websocket;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonObject;
import com.one2manycall.vo.UserSession;
import org.kurento.client.EventListener;
import org.kurento.client.IceCandidate;
import org.kurento.client.IceCandidateFoundEvent;
import org.kurento.client.WebRtcEndpoint;
import org.kurento.jsonrpc.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;


import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;

import org.kurento.client.KurentoClient;
import org.kurento.client.MediaPipeline;

/**
 * 接受websocket 消息
 */
public class CallHandler extends TextWebSocketHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(CallHandler.class);


    private static final Gson gson = new GsonBuilder().create();


    /**
     * 所有的观看者
     */
    private final ConcurrentHashMap<String, UserSession> viewers = new ConcurrentHashMap<>();

    @Autowired
    private KurentoClient kurento;

    @Autowired
    private PlayerHandler playerHandler;

    private MediaPipeline pipeline;

    /**
     * 直播者 的session
     */
    private UserSession presenterUserSession;


    @Autowired
    private UserRegistry registry;

    @Autowired
    private MessageHandler messageHandler;

    /**
     * 消息处理类
     *
     * @param session session
     * @param message message
     * @throws Exception
     */



    @Autowired
    private VideoHandler videoHandler;

    @Override
    public void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        JsonObject jsonMessage = gson.fromJson(message.getPayload(), JsonObject.class);
        LOGGER.info("Incoming message from session '{}': {}", session.getId(), jsonMessage);


//        String type = jsonMessage.get("type").getAsString();
//
//
//        // 如果是播放视频 直接这儿处理
//        if (type !=null){
//            videoHandler.handleTextMessage(session,message);
//            return;
//        }


        // 如果是播放视频 直接这儿处理
//        if ("play".equals(type)){
//            playerHandler.handleTextMessage(session,message);
//            return;
//        }

        UserSession userFromSession = registry.getBySession(session);

        if (userFromSession != null) {
            LOGGER.debug("Incoming message from user '{}': {}", userFromSession.getName(), jsonMessage);
        } else {
            LOGGER.debug("Incoming message from new user: {}", jsonMessage);
        }


        switch (jsonMessage.get("id").getAsString()) {

            case "chat":
                messageHandler.chat(session, jsonMessage);
                break;

            case "start":
                messageHandler.start(session, jsonMessage);
                break;

            case "register":
                messageHandler.registerTextHander(session, jsonMessage);
                break;

            // 表演者
            case "presenter":
                try {
                    presenter(session, jsonMessage);
                } catch (Throwable t) {
                    messageHandler.handleErrorResponse(t, session, "presenterResponse");
                }
                break;
            // 观看者
            case "viewer":
                try {
                    viewer(session, jsonMessage);
                } catch (Throwable t) {
                    messageHandler.handleErrorResponse(t, session, "viewerResponse");
                }
                break;

            case "play":
                messageHandler.play(userFromSession, jsonMessage);
                break;

            // 网络信息收集
            case "onIceCandidate": {
                JsonObject candidate = jsonMessage.get("candidate").getAsJsonObject();

                UserSession user = null;
                if (presenterUserSession != null) {
                    if (presenterUserSession.getSession() == session) {
                        user = presenterUserSession;
                    } else {
                        user = viewers.get(session.getId());
                    }
                }
                if (user != null) {
                    IceCandidate cand =
                            new IceCandidate(candidate.get("candidate").getAsString(), candidate.get("sdpMid")
                                    .getAsString(), candidate.get("sdpMLineIndex").getAsInt());
                    user.addCandidate(cand);
                }
                break;
            }

            // 停止访问
            case "stop":
                stop(session);
                break;
            default:
                break;
        }
    }



    private synchronized void presenter(final WebSocketSession session, JsonObject jsonMessage)
            throws IOException {

        LOGGER.info("presenter !!! ");
        if (presenterUserSession == null) {
            presenterUserSession = new UserSession(session);

            pipeline = kurento.createMediaPipeline();
            presenterUserSession.setWebRtcEndpoint(new WebRtcEndpoint.Builder(pipeline).build());

            WebRtcEndpoint presenterWebRtc = presenterUserSession.getWebRtcEndpoint();

            presenterWebRtc.addIceCandidateFoundListener(new EventListener<IceCandidateFoundEvent>() {

                @Override
                public void onEvent(IceCandidateFoundEvent event) {
                    JsonObject response = new JsonObject();
                    response.addProperty("id", "iceCandidate");
                    response.add("candidate", JsonUtils.toJsonObject(event.getCandidate()));
                    try {
                        synchronized (session) {
                            session.sendMessage(new TextMessage(response.toString()));
                        }
                    } catch (IOException e) {
                        LOGGER.debug(e.getMessage());
                    }
                }
            });

            String sdpOffer = jsonMessage.getAsJsonPrimitive("sdpOffer").getAsString();
            String sdpAnswer = presenterWebRtc.processOffer(sdpOffer);

            JsonObject response = new JsonObject();
            response.addProperty("id", "presenterResponse");
            response.addProperty("response", "accepted");
            response.addProperty("sdpAnswer", sdpAnswer);

            synchronized (session) {
                presenterUserSession.sendMessage(response);
            }
            presenterWebRtc.gatherCandidates();

        } else {
            JsonObject response = new JsonObject();
            response.addProperty("id", "presenterResponse");
            response.addProperty("response", "rejected");
            response.addProperty("message",
                    "Another user is currently acting as sender. Try again later ...");
            session.sendMessage(new TextMessage(response.toString()));
        }
    }

    private synchronized void viewer(final WebSocketSession session, JsonObject jsonMessage)
            throws IOException {

        LOGGER.info("view registry start: " + session.getId());
        // 当 表演者还未准备好的情况下，不能观看，给返回结果
        if (presenterUserSession == null || presenterUserSession.getWebRtcEndpoint() == null) {
            JsonObject response = new JsonObject();
            response.addProperty("id", "viewerResponse");
            response.addProperty("response", "rejected");
            response.addProperty("message",
                    "No active sender now. Become sender or . Try again later ...");
            session.sendMessage(new TextMessage(response.toString()));

            return;
        }
        // 如果已经在观看列表里面，返回提示内容
        if (viewers.containsKey(session.getId())) {
            JsonObject response = new JsonObject();
            response.addProperty("id", "viewerResponse");
            response.addProperty("response", "rejected");
            response.addProperty("message", "You are already viewing in this session. "
                    + "Use a different browser to add additional viewers.");
            session.sendMessage(new TextMessage(response.toString()));
            return;
        }
        UserSession viewer = new UserSession(session);

        // 添加心的观看人员
        viewers.put(session.getId(), viewer);

        WebRtcEndpoint nextWebRtc = new WebRtcEndpoint.Builder(pipeline).build();

        nextWebRtc.addIceCandidateFoundListener(new EventListener<IceCandidateFoundEvent>() {

            @Override
            public void onEvent(IceCandidateFoundEvent event) {
                JsonObject response = new JsonObject();
                response.addProperty("id", "iceCandidate");
                response.add("candidate", JsonUtils.toJsonObject(event.getCandidate()));
                try {
                    synchronized (session) {
                        session.sendMessage(new TextMessage(response.toString()));
                    }
                } catch (IOException e) {
                    LOGGER.debug(e.getMessage());
                }
            }
        });

        viewer.setWebRtcEndpoint(nextWebRtc);
        presenterUserSession.getWebRtcEndpoint().connect(nextWebRtc);
        String sdpOffer = jsonMessage.getAsJsonPrimitive("sdpOffer").getAsString();
        String sdpAnswer = nextWebRtc.processOffer(sdpOffer);

        JsonObject response = new JsonObject();
        response.addProperty("id", "viewerResponse");
        response.addProperty("response", "accepted");
        response.addProperty("sdpAnswer", sdpAnswer);

        synchronized (session) {
            viewer.sendMessage(response);
        }
        nextWebRtc.gatherCandidates();

    }

    /**
     * 结束会话
     * @param session session
     * @throws IOException
     */

    private synchronized void stop(WebSocketSession session) throws IOException {

        LOGGER.info("stop session : " + session.getId());
        // 表演者 会话结束处理
        String sessionId = session.getId();
        if (presenterUserSession != null && presenterUserSession.getSession().getId().equals(sessionId)) {
            // 循环给每一个观看者发送 断开视频的消息
            for (UserSession viewer : viewers.values()) {
                JsonObject response = new JsonObject();
                response.addProperty("id", "stopCommunication");
                viewer.sendMessage(response);
            }

            LOGGER.info("Releasing media pipeline");
            if (pipeline != null) {
                pipeline.release();
            }
            pipeline = null;
            presenterUserSession = null;
            return;
        }
        // 观看者 处理
        if (viewers.containsKey(sessionId)) {

            LOGGER.info("stop viewer  : " +sessionId);

            // 从观看列表剔除 这个sessionId
            if (viewers.get(sessionId).getWebRtcEndpoint() != null) {
                viewers.get(sessionId).getWebRtcEndpoint().release();
            }
            viewers.remove(sessionId);
            return;
        }

        LOGGER.error("other session : " + sessionId);
    }

    /**
     *  监听连接是否断开
     * @param session se
     * @param status status
     * @throws Exception
     */

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        stop(session);
    }

}
